package com.yanggu.flink.datastream_api.source

import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import java.util.concurrent.TimeUnit
import scala.util.Random

/**
 * 自定义Source, 实现SourceFunction接口, 重写run()方法和cancel()方法即可
 * SourceFunction的并行度只能设置为1, 如果要自定义并行的Source需要实现ParallelSourceFunction
 */
object SourceFromMySource {

  def main(args: Array[String]): Unit = {
    //创建环境对象
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    //添加自定义的source
    //设置成多并行度会直接报错
    val streaming = environment.addSource(new MySensorSource)/*.setParallelism(2)*/

    //打印, 并且设置并行度为1
    streaming.print().setParallelism(1)

    //开启任务
    environment.execute()

  }

}
class MySensorSource extends SourceFunction[SensorReading] {

  // flag: 表示数据源是否还在运行
  var running = true

  override def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {

    val random = new Random

    while (running) {
      val curTime = System.currentTimeMillis()
      1.to(5).foreach(
        //使用ctx对象（上下文）向下游发送数据
        i => ctx.collect(SensorReading("sensor_" + i, curTime, 65 + random.nextGaussian() * 20))
      )
      TimeUnit.MILLISECONDS.sleep(100L)
    }
  }

  override def cancel(): Unit = running = false

}
